#include "mwm_index.h"


//int pthread_setname_np(const char *name);
//#define fz_set_thread_title(name) pthread_setname_np(name)


TCLIST *fz_res_list[IO_THREADS_MAX_NUM];
task_args_t g_task[IO_THREADS_MAX_NUM];
pthread_t fz_threads[IO_THREADS_MAX_NUM];
pthread_mutex_t fz_threads_mutex[IO_THREADS_MAX_NUM];
int  g_init_thread = 0;
extern pthread_mutex_t g_ret_mutex;


void close_table_key( RedisModuleKey* HashKey);
int all_search_02(char *search_data, int real_len, void *data);
bool is_already_hit_clue_thread_03(RedisModuleCtx *ctx, RedisModuleKey* jq, RedisModuleString *key, char *value , size_t len, mwm_ctx_t_thread* mwmctx, int s, TCMAP *tmpMap ,TCMAP *resMap);


void makeThreadKillable(void) {
;
    //pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
    //pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);
}

void proc_segment_data(task_args_t task, TCLIST *res, TCMAP *resMap, TCMAP *tmpMap, TCMAP *hitsResMap){

	RedisModuleCtx *ctx = task.ctx;
	//RedisModuleString **argv = task.argv;
	char **argv = task.argv;

	mwm_ctx_t_thread mhctx;
	mhctx.ctx = ctx;
	mhctx.HashKey = task.mh;
    
	int i  = 0;
	int pre_num = task.start;
	int ret_line = task.line;
	for (i = 0; i < task.argc; i++) 
	{
		size_t size_len_value = 0;

		//pthread_mutex_lock(&g_ret_mutex);
		//char *value = RM_StringPtrLen(argv[i], &size_len_value);
		char *value = argv[i];
		//char *value = (char *)((robj *)argv[i])->ptr;
		size_len_value = strlen(value);
		//pthread_mutex_unlock(&g_ret_mutex);
		if(strncmp(value, "F#Z", strlen("F#Z")) == 0)
		{
			res_return_client_05(ctx, pre_num, i, ret_line, task.index, NULL, tmpMap, resMap, hitsResMap, res);
			tcmapclear(hitsResMap);
			tcmapclear(resMap);
			tcmapclear(tmpMap);
			pre_num = i + 1;
			ret_line++;
			continue;
		}
		RedisModuleString* key = RM_CreateString( NULL, value, size_len_value );

		if(!is_already_hit_clue_thread_03(ctx, task.jq, key, value, size_len_value, &mhctx, i, tmpMap, resMap))
		{
			all_search_02(value, size_len_value, &mhctx);
		}	
		RedisModule_FreeString(NULL,key);
	}
}


void *fzThreadMain(void *myid) {
    /* The ID is the thread number (from 0 to server.iothreads_num-1), and is
     * used by the thread to just manipulate a single sub-array of clients. */
    long id = (unsigned long)myid;
    char thdname[16];
    snprintf(thdname, sizeof(thdname), "FZ_GET_thd_%ld", id);
    //fz_set_thread_title(thdname);
    //makeThreadKillable();
	TCMAP *resMap = tcmapnew();
	TCMAP *tmpMap = tcmapnew();
	TCMAP *hitsResMap = tcmapnew();
	
    while(1) {
        // process 
        //sleep(10);
        while(g_task[id].use  == -1 || g_task[id].use  == 0)
        {
        	usleep(20);
        }

        //RedisModule_ThreadSafeContextLock(g_task[id].ctx);
        RM_Log(g_task[id].ctx, "warning", "############");

        tclistclear(fz_res_list[id]);
	    //task_args_t task =  g_task[id];
		proc_segment_data(g_task[id], fz_res_list[id], resMap, tmpMap, hitsResMap);  
		//RedisModule_ThreadSafeContextUnlock(g_task[id].ctx);
        tcmapclear(resMap);
        tcmapclear(tmpMap);
        tcmapclear(hitsResMap);
        g_task[id].use = 0;
    }
}

void Init_thread()
{
    for (int i = 0; i < IO_THREADS_MAX_NUM; i++) 
    {        
	    /* Things we do for all the threads including the main thread. */ 
	    fz_res_list[i] = tclistnew();   
	    g_task[i].use = -1;      
	    /* Things we do only for the additional threads. */        
	    pthread_t tid;        
	    pthread_mutex_init(&fz_threads_mutex[i],NULL);        
	    /* Thread will be stopped. */        
	    if (pthread_create(&tid, NULL, fzThreadMain,(void*)(long)i) != 0) 
	    {            
	    	exit(1);        
	    }       
	    fz_threads[i] = tid;    
    }
    g_init_thread = 1;
}


